package com.lzq.mq.nameserver.replication;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.SlaveAckDTO;
import com.lzq.mq.common.enums.NameServerEventCode;
import com.lzq.mq.common.enums.NameServerResponseCode;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.common.MasterSlaveReplicationProperties;
import com.lzq.mq.nameserver.enums.MasterSlaveReplicationTypeEnum;
import com.lzq.mq.nameserver.event.model.ReplicationMsgEvent;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class MasterReplicationMsgSendTask extends ReplicationTask {

    private BlockingQueue<ReplicationMsgEvent> replicationMsgQueue = new ArrayBlockingQueue(5000);

    public MasterReplicationMsgSendTask(String taskName) {
        super(taskName);
    }


    @Override
    void startTask() {
        log.info("开始启动主节点数据同步任务...");
        MasterSlaveReplicationProperties masterSlaveReplicationProperties = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties();
        MasterSlaveReplicationTypeEnum replicationTypeEnum = MasterSlaveReplicationTypeEnum.of(masterSlaveReplicationProperties.getType());
        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent =
                    CommonCache.getReplicationMsgQueueManager().getReplicationMsgQueue().take();    // 获取需要同步的实例
                replicationMsgEvent.setMsgId(UUID.randomUUID().toString());                         // 设置id
                Channel brokerChannel = replicationMsgEvent.getChannelHandlerContext().channel();

                Map<String, ChannelHandlerContext> channelHandlerContextMap = CommonCache.getReplicationChannelManager().getValidSlaveChannelMap();
                int validSlaveChannelCount = channelHandlerContextMap.size();

                if (replicationTypeEnum == MasterSlaveReplicationTypeEnum.ASYNC) {
                    this.sendMsgToSlave(replicationMsgEvent);
                    brokerChannel.writeAndFlush(
                            new TcpMsg(
                                    NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getCode(),
                                    NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getDesc().getBytes()
                            )
                    );
                } else if (replicationTypeEnum == MasterSlaveReplicationTypeEnum.SYNC) {
                    //需要接收到多少个ack的次数
                    this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount);
                    this.sendMsgToSlave(replicationMsgEvent);
                } else if (replicationTypeEnum == MasterSlaveReplicationTypeEnum.HALF_SYNC) {
                    this.inputMsgToAckMap(replicationMsgEvent, validSlaveChannelCount / 2);
                    this.sendMsgToSlave(replicationMsgEvent);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 将主节点需要发送出去的数据注入到一个map中，然后当从节点返回ack的时候，该map的数据会被剔除对应记录
     * @param replicationMsgEvent 需要同步的节点信息
     * @param needAckCount  需要ack的次数
     */
    // todo 存储需要ack的次数, 那要是一个节点ack两次不就炸缸了?
    // 考虑使用set存储需要ack的子服务id, 每次ack剔除一个元素即可
    private void inputMsgToAckMap(ReplicationMsgEvent replicationMsgEvent, int needAckCount) {
        CommonCache.getAckMap().put(
                replicationMsgEvent.getMsgId(),
                new SlaveAckDTO(new AtomicInteger(needAckCount), replicationMsgEvent.getChannelHandlerContext())
        );
    }

    /**
     * 将broker节点信息同步给slave-nameserver
     */
    private void sendMsgToSlave(ReplicationMsgEvent replicationMsgEvent) {
        Map<String, ChannelHandlerContext> channelHandlerContextMap = CommonCache.getReplicationChannelManager().getValidSlaveChannelMap();
        //判断当前采用的同步模式是哪种方式
        for (String reqId : channelHandlerContextMap.keySet()) {
            replicationMsgEvent.setChannelHandlerContext(null);
            byte[] body = JSON.toJSONBytes(replicationMsgEvent);        // 将节点信息转换成json
            //异步复制，直接发送给从节点，然后告知broker注册成功
            TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.MASTER_REPLICATION_MSG.getCode(), body);
            channelHandlerContextMap.get(reqId).writeAndFlush(tcpMsg);
        }
    }
}
